//
// Created by 郝羽 on 2018/5/22.
//

#include "remote_mixer.hpp"

#include <gio/gio.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <list>

#include "remote_address_list.hpp"
#include "rtp.hpp"
#include "xm_log.h"
#include "xm_app_config.h"
#include "xcutil.h"

constexpr int stream_timeout = 10000;
constexpr int heartbeat_interval = 1000;
const char* heartbeat = "HEARTBEAT";

struct rmixer_stream {
    sockaddr_in sa;
    int64_t last_recv_time;
};

struct rmixer {
    char* mixer_id;

    GSocket* sock;

    guint sock_in;

    rmixer_callback callback;

    void* callback_data;

    remote_address_list* address_list;

    std::map<uint32_t, rmixer_stream> streams;

    int64_t last_heartbeat_time;
};

static gboolean _rmixer_handle_socket_recv(GIOChannel *source, GIOCondition conf, gpointer data);
static gboolean _rmixer_handle_socket_recv_rtp(rmixer* mixer, sockaddr_in sa, const unsigned char* buf, int size);
static gboolean _rmixer_handle_socket_recv_rtcp(rmixer* mixer, sockaddr_in sa, const unsigned char* buf, int size);

rmixer *rmixer_new(const char* mixer_id, rmixer_callback callback, void *data) {
    GError* err = nullptr;

    auto sock = g_socket_new(G_SOCKET_FAMILY_IPV4, G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err);

    if (err != nullptr) {
        dbge("remote mixer '%s' could not create remote mixer socket, error: %s",
             mixer_id, err->message);

        g_error_free(err);

        return nullptr;
    }

    g_socket_bind(sock, g_inet_socket_address_new_from_string(app_config_get()->ipm_ip, 0), FALSE, &err);

    if (err != nullptr) {
        dbge("remote mixer '%s' could not bind remote mixer socket, error: %s",
             mixer_id, err->message);

        g_error_free(err);

        g_object_unref(sock);

        return nullptr;
    }

    auto local_addr = g_socket_get_local_address(sock, &err);

    if (err != nullptr) {
        dbge("remote mixer '%s' could not get remote mixer local address, error: %s",
             mixer_id, err->message);

        g_error_free(err);

        g_object_unref(sock);

        return nullptr;
    }

    sockaddr_in local_addr_sa;

    g_socket_address_to_native(local_addr, &local_addr_sa, sizeof(sockaddr_in), &err);

    g_object_unref(local_addr);

    if (err != nullptr) {
        dbge("remote mixer '%s' could not convert remote mixer local address to native, error: %s",
             mixer_id, err->message);

        g_error_free(err);

        g_object_unref(sock);

        return nullptr;
    }

    dbgi("remote mixer '%s' bind to '%s:%d'",
         mixer_id, inet_ntoa(local_addr_sa.sin_addr), ntohs(local_addr_sa.sin_port));

    auto mixer = new rmixer();
    mixer->mixer_id = g_strdup(mixer_id);
    mixer->sock = sock;
    mixer->callback = callback;
    mixer->callback_data = data;
    mixer->last_heartbeat_time = 0;

    auto ch = g_io_channel_unix_new(g_socket_get_fd(sock));

    mixer->sock_in = g_io_add_watch(ch, G_IO_IN, _rmixer_handle_socket_recv, mixer);

    mixer->address_list = remote_address_list_new();

    return mixer;
}

static gboolean _rmixer_handle_socket_recv(GIOChannel *source, GIOCondition conf, gpointer data) {
    auto mixer = static_cast<rmixer*>(data);

    unsigned char buf[1600] = {0,};

    GError* err = nullptr;

    int size = 0;

    GSocketAddress* addr = nullptr;

    size = g_socket_receive_from(mixer->sock, &addr, (gchar*)buf, sizeof(buf), nullptr, &err);

    if (err != nullptr) {
        dbge("remote mixer '%s' could not receive from socket: %s",
             mixer->mixer_id,
             err->message);

        g_error_free(err);
        return G_SOURCE_CONTINUE;
    }

    sockaddr_in sa;

    g_socket_address_to_native(addr, &sa, sizeof(sa), &err);

    if (err != nullptr) {
        dbge("remote mixer '%s' could not convert address to native: %s",
             mixer->mixer_id,
             err->message);

        g_error_free(err);
        g_object_unref(addr);

        return G_SOURCE_CONTINUE;
    }

    remote_address_list_update(mixer->address_list, sa);

    if ((size_t)size == strlen(heartbeat) && memcmp(buf, heartbeat, strlen(heartbeat)) == 0) {
        // dbgi("remote mixer %s got HEARTBEAT from %s:%d", mixer->mixer_id, inet_ntoa(sa.sin_addr), ntohs(sa.sin_port));
    } else if (is_rtcp(buf, size)) {
        _rmixer_handle_socket_recv_rtcp(mixer, sa, buf, size);
    } else {
        _rmixer_handle_socket_recv_rtp(mixer, sa, buf, size);
    }

    return G_SOURCE_CONTINUE;
}

static gboolean _rmixer_handle_socket_recv_rtcp(rmixer* mixer, sockaddr_in sa, const unsigned char* buf, int size) {
    return G_SOURCE_CONTINUE;
}

static gboolean _rmixer_handle_socket_recv_rtp(rmixer* mixer, sockaddr_in sa, const unsigned char* buf, int size) {
    uint32_t ssrc = rtp_get_ssrc(buf, size);

    auto now = get_timestamp_ms();

    auto it = mixer->streams.find(ssrc);

    if (it == mixer->streams.end()) {
        mixer->callback.on_stream_new(mixer->callback_data, ssrc);

        mixer->streams.emplace(ssrc, rmixer_stream{ .sa = sa, .last_recv_time = now });
    } else {
        mixer->streams[ssrc].last_recv_time = now;
    }

    mixer->callback.on_stream_data(mixer->callback_data, ssrc, buf, size);

    return G_SOURCE_CONTINUE;
}

void _rmixer_handle_remote_address_destroy(void* data, sockaddr_in sa) {
    // auto mixer = static_cast<rmixer*>(data);
}

void rmixer_del(rmixer *mixer) {
    GError* err = nullptr;

    if (mixer->sock_in > 0) {
        g_source_remove(mixer->sock_in);
    }

    g_socket_close(mixer->sock, &err);

    if (err != nullptr) {
        dbge("remote mixer '%s' could not close remote mixer socket: %s",
             mixer->mixer_id,
             err->message);

        g_error_free(err);
    }

    g_object_unref(mixer->sock);

    remote_address_list_del(mixer->address_list);

    g_free(mixer->mixer_id);

    delete(mixer);
}

const char* rmixer_get_local_address(rmixer *mixer) {
    GError* err = nullptr;

    auto addr = g_socket_get_local_address(mixer->sock, &err);

    if (err != nullptr) {
        dbge("remote mixer '%s' could not get local address: %s",
             mixer->mixer_id,
             err->message);

        g_error_free(err);

        return nullptr;
    }

    struct sockaddr_in sa;

    g_socket_address_to_native(addr, &sa, sizeof(sa), &err);

    g_object_unref(addr);

    if (err != nullptr) {
        dbge("remote mixer '%s' could not convert local address to native: %s",
             mixer->mixer_id,
             err->message);

        g_error_free(err);

        return nullptr;
    }

    return g_strdup_printf("%s:%d", inet_ntoa(sa.sin_addr), ntohs(sa.sin_port));
}

void rmixer_send_data(rmixer *mixer, const unsigned char *buf, int size) {
    remote_address_list_each(mixer->address_list, [&](sockaddr_in sa){
        auto gsa = g_socket_address_new_from_native(&sa, sizeof(sockaddr_in));
        g_socket_send_to(mixer->sock, gsa, (gchar*)buf, size, nullptr, nullptr);
    });
}

void rmixer_handle_timeout(rmixer *mixer, int64_t now) {
    auto it = mixer->streams.begin();
    while (it != mixer->streams.end()) {
        auto cur = it++;
        auto ssrc = cur->first;
        auto stream = cur->second;

        if (now - stream.last_recv_time > stream_timeout) {
            mixer->callback.on_stream_destroy(mixer->callback_data, ssrc);
            mixer->streams.erase(cur);
        }
    }

    if (now - mixer->last_heartbeat_time > heartbeat_interval) {
        rmixer_send_data(mixer, (const unsigned char*)heartbeat, strlen(heartbeat));
        mixer->last_heartbeat_time = now;
    }

    remote_address_list_handle_timeout(mixer->address_list, now);
}

int rmixer_join(rmixer *mixer, sockaddr_in sa) {
    remote_address_list_update(mixer->address_list, sa);
    return 0;
}
